package com.leijie.grpc;

import com.leijie.proto.ListResponse;
import com.leijie.proto.PersionServiceGrpc;
import com.leijie.proto.StrResponse;
import com.leijie.proto.StreamRequest;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import io.grpc.stub.StreamObserver;

import java.time.LocalDate;
import java.util.Iterator;

/**
 * @author LeiJie
 * @time 2020-5-7 14:29
 * @desc TODO
 */
public class GrpcClient {

    public static void main(String[] args) throws InterruptedException {
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost",8899).usePlaintext().build();
        PersionServiceGrpc.PersionServiceBlockingStub persionServiceStub = PersionServiceGrpc.newBlockingStub(channel);

        com.leijie.proto.MyResponse response = persionServiceStub.getRealNameByUsername(com.leijie.proto.MyRequest.newBuilder().setUsername("测试").build());
        System.out.println("服务端返回值：>>>"+response.getRealname());

        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>");

        Iterator<com.leijie.proto.StreamResponse> it = persionServiceStub.getPersionByAge(com.leijie.proto.StreamRequest.newBuilder().setAge(10000).build());
        while (it.hasNext()){
            com.leijie.proto.StreamResponse  rs = it.next();
            System.out.println(rs.getName()+","+rs.getAge()+","+rs.getCity());
        }

        System.out.println(">>>>>>>>>>>>>>>>>>>>>>>");

        //流式参数为异步方式接收数据
        PersionServiceGrpc.PersionServiceStub stub = PersionServiceGrpc.newStub(channel);

        StreamObserver<ListResponse> listResponseStreamObserver = new StreamObserver<ListResponse>() {
            @Override
            public void onNext(ListResponse value) {
                System.out.println(">>>>>收到服务端返回数据<<<<<<<<<<<<");
                value.getStreamResponseList().forEach(rs->{
                    System.out.println(rs.getName()+","+rs.getAge()+","+rs.getCity());
                });
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                System.out.println("completed!....");
            }
        };
        StreamObserver<StreamRequest> rs = stub.getPersionWrapperByAges(listResponseStreamObserver);
        rs.onNext(com.leijie.proto.StreamRequest.newBuilder().setAge(1000).build());
        rs.onNext(com.leijie.proto.StreamRequest.newBuilder().setAge(2000).build());

        rs.onCompleted();


        StreamObserver<com.leijie.proto.StrRequest> streamObserver = stub.biTalk(new StreamObserver<StrResponse>() {
            @Override
            public void onNext(StrResponse value) {
                System.out.println("收到服务器返回数据：>>>"+value.getResponseInfo());
            }

            @Override
            public void onError(Throwable t) {
                t.printStackTrace();
            }

            @Override
            public void onCompleted() {
                System.out.println("completed!....");
            }
        });

        for (int i=0;i<10;i++){
            streamObserver.onNext(com.leijie.proto.StrRequest.newBuilder().setRequestInfo(LocalDate.now().toString()).build());
        }
        streamObserver.onCompleted();


        Thread.sleep(10000);
    }
}
